package main

import (
	"time"
	"github.com/sirupsen/logrus"
	"os"
	"bufio"
	"strings"
	"github.com/mgutz/str"
	"net/url"
	"crypto/md5"
	"encoding/hex"
	"strconv"
	"github.com/mediocregopher/radix.v2/pool"
)

const HANDLE_TAG = "/dig?"
const HANDLE_LIST = "/list/"
const HANDLE_MOVIE = "/movie/"
const HANDLE_HTML = ".html"

type urlData struct {
	data  digData
	uid   string
	unode urlNode
}

type digData struct {
	time  string
	url   string
	ua    string
	refer string
}

type cmdParams struct {
	logPath    string
	routineNum int
}
type urlNode struct {
	unType string //  / /movie  /list
	unRid  int
	unUrl  string
	unTime string
}
type storageBlock struct {
	counterType string
	model       string
	unode       urlNode
}

var log = logrus.New()

func init() {
	log.Out = os.Stdout
	log.SetLevel(logrus.DebugLevel)
}

func main() {
	logPath := "/Users/shuchang/Documents/nginx/logs/digs.log"
	routineNum := 10
	targetPath := "runtime.log"

	file, e := os.OpenFile(targetPath, os.O_CREATE|os.O_RDWR, 0644)
	if e == nil {
		defer file.Close()
		log.Out = file
	}
	log.Infoln("EXEC START")
	params := cmdParams{
		logPath:    logPath,
		routineNum: routineNum,
	}

	pool, err := pool.New("tcp", "localhost:6379", params.routineNum*2)
	if err != nil {
		panic(err)
	} else {
		go func() {
			for {
				pool.Cmd("PING")
				time.Sleep(3 * time.Second)
			}
		}()
	}

	logChan := make(chan string, 3*routineNum)
	pvChan := make(chan urlData, routineNum)
	uvChan := make(chan urlData, routineNum)
	storageChan := make(chan storageBlock, routineNum)

	go readLogFile(params, logChan)

	for i := 0; i < params.routineNum; i++ {
		go logConsumer(logChan, pvChan, uvChan)
	}

	go pvCounter(pvChan, storageChan)
	go uvCounter(uvChan, storageChan, pool)

	go store(storageChan, pool)

	time.Sleep(1000 * time.Second)
}

func store(blocks chan storageBlock, redisPool *pool.Pool) {
	for block := range blocks {
		prefix := block.counterType + "_"
		//逐层添加
		//维度 天 小时 分组
		//层级 顶级 大分类  小分类 终极
		setKeys := []string{
			prefix + "day_" + getTime(block.unode.unTime, "day"),
			prefix + "hour_" + getTime(block.unode.unTime, "hour"),
			prefix + "min_" + getTime(block.unode.unTime, "min"),
			prefix + block.unode.unType + "_day_" + getTime(block.unode.unTime, "day"),
			prefix + block.unode.unType + "_hour_" + getTime(block.unode.unTime, "hour"),
			prefix + block.unode.unType + "_min _" + getTime(block.unode.unTime, "min"),
		}
		rowId := block.unode.unRid
		for _, key := range setKeys {
			ret, e := redisPool.Cmd(block.model, key, 1, rowId).Int()
			if ret <= 0 || e != nil {
				log.Errorln("存储block 出错 ", e)
			}
		}
	}
}

func uvCounter(uvchan chan urlData, blocks chan storageBlock, redisPool *pool.Pool) {
	for data := range uvchan {
		//hyperloglog  redis
		hyperloglogKey := "uv_" + getTime(data.unode.unTime, "day")
		ret, e := redisPool.Cmd("PFADD", hyperloglogKey, data.uid, "EX", 86400).Int()
		if e != nil {
			logrus.Warningln("redis 操作 ", e)
		}
		if ret != 1 {
			continue
		}
		block := storageBlock{
			"uv", "ZINCRBY", data.unode,
		}
		blocks <- block
	}
}
func pvCounter(pvchan chan urlData, blocks chan storageBlock) {
	for data := range pvchan {
		block := storageBlock{
			"pv", "ZINCRBY", data.unode,
		}
		blocks <- block
	}
}

func logConsumer(logchan chan string, pvchan chan urlData, uvChan chan urlData) error {
	for logStr := range logchan {
		data := cutLog(logStr)
		//uid 模拟生成 md5(refer+ua)
		hash := md5.New()
		hash.Write([]byte(data.ua + data.refer))
		uid := hex.EncodeToString(hash.Sum(nil))

		udata := urlData{
			data:  data,
			uid:   uid,
			unode: formatUrl(data.url, data.time),
		}
		logrus.Infoln(udata)
		pvchan <- udata
		uvChan <- udata
	}
	return nil
}
func cutLog(logstr string) digData {
	logstr = strings.TrimSpace(logstr)
	pos1 := str.IndexOf(logstr, HANDLE_TAG, 0)
	if pos1 == -1 {
		return digData{}
	}
	pos1 += len(HANDLE_TAG)
	pos2 := str.IndexOf(logstr, "HTTP/", pos1)
	d := str.Substr(logstr, pos1, pos2-pos1)

	parse, e := url.Parse("http://localhost/?" + d)

	if e != nil {
		return digData{}
	}
	data := parse.Query()
	return digData{
		time:  data.Get("time"),
		refer: data.Get("refer"),
		url:   data.Get("url"),
		ua:    data.Get("ua"),
	}
}

func readLogFile(params cmdParams, logchan chan string) {
	file, e := os.OpenFile(params.logPath, os.O_CREATE|os.O_RDWR, 0644)
	if e == nil {
		defer file.Close()
		log.Out = file
	}
	count := 0
	logrus.Infoln("读取日志文件")
	reader := bufio.NewReader(file)
	for {
		line, err := reader.ReadString('\n')
		if err != nil {
			panic(err)
		}
		logchan <- line
		count++
		if count%(1000*params.routineNum) == 0 {
			logrus.Infoln("读取%d行", count)
		}
	}
}

func formatUrl(url string, time string) urlNode {
	pos1 := str.IndexOf(url, HANDLE_MOVIE, 0)
	if pos1 != -1 {
		pos1 += len(HANDLE_MOVIE)
		pos2 := str.IndexOf(url, HANDLE_HTML, pos1)
		d := str.Substr(url, pos1, pos2-pos1)
		id, _ := strconv.Atoi(d)
		return urlNode{
			"movie", id, url, time,}
	} else {
		pos1 = str.IndexOf(url, HANDLE_LIST, 0)
		if pos1 != -1 {
			pos1 += len(HANDLE_LIST)
			pos2 := str.IndexOf(url, HANDLE_HTML, pos1)
			d := str.Substr(url, pos1, pos2-pos1)
			id, _ := strconv.Atoi(d)
			return urlNode{
				"list", id, url, time,}
		}
	}
	return urlNode{"home", 1, url, time}
}

func getTime(logtime, timetype string) string {
	var item string
	switch timetype {
	case "day":
		item = "2006-01-02"
		break
	case "hour":
		item = "2006-01-02 15"
		break
	case "min":
		item = "2006-01-02 15:04"
		break
	}
	t, _ := time.Parse(item, time.Now().Format(item))
	return strconv.FormatInt(t.Unix(), 10)
}
